

package finalProject;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.Reducer;

public class Step1 {

	private static int dpMin = 1;
	private static boolean withInvert = false;

	public static class MapClass extends
	Mapper<LongWritable, Text, Path, NounPair> {

		
		@Override
		protected void setup(Context context)
				throws IOException, InterruptedException {

			withInvert = context.getConfiguration().get("withInvert").equals("inver");
		}

		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {

			String[] parts = value.toString().split("\t");
			if (parts.length < 2) {
				return;
			}
			ArrayList<Token> tokens = createTokens(parts[1].split(" "));
			if (countNouns(tokens) < 2) {
				return;
			}

			Tree t = new Tree(tokens);

			for(int sIndex = 0; sIndex < tokens.size() ; sIndex++) {
				Token source = tokens.get(sIndex);
				if(source.isNoun() == true)
				{
					for(int tIndex = sIndex + 1 ; tIndex < tokens.size() ; tIndex++)
					{
						Token target = tokens.get(tIndex);
						if(target.isNoun() == true)
						{
							String path = t.getPath(source, target);
							if (path == null)
							{
								continue;
							}

							NounPair nounObj = new NounPair(source.getWord(), target.getWord());
							Path pathObj = new Path(new Text(path), nounObj);
							context.write(pathObj, nounObj);
							if(withInvert == true)
							{
								path = t.getPath(target, source);
								nounObj = new NounPair(target.getWord(), source.getWord());
								pathObj = new Path(new Text("@@@R@@" + path), nounObj);
								context.write(pathObj, nounObj);
							}
						}
					}
				}
			}

		}

		private ArrayList<Token> createTokens(String[] parts) {
			ArrayList<Token> ans = new ArrayList<Token>();
			for (String string : parts) {
				ans.add(new Token(string));
			}
			return ans;
		}

		private int countNouns(ArrayList<Token> tokens) {
			int nounCounter = 0;
			for (Token token : tokens) {
				if (token.isNoun())
					nounCounter++;
			}
			return nounCounter;
		}
	}

	public static class PartitionClass extends Partitioner<Path, NounPair> {

		@Override
		public int getPartition(Path key, NounPair value, int partitionNum) {

			int ans = key.getPath().toString().hashCode() % partitionNum;
			while (ans < 0) {
				ans += partitionNum;
			}
			return ans;
		}
	}
	
	//Grouping Comparator
	public static class PathComparatorByPath extends WritableComparator {

		public PathComparatorByPath () {
			super(Path.class, true);
		}

		@SuppressWarnings("rawtypes")
		@Override
		public int compare(WritableComparable a, WritableComparable b) {
			Path p1 = (Path) a;
			Path p2 = (Path) b;
			return p1.getPath().toString().compareTo(p2.getPath().toString());
		}
	}

	// Sorter
	public static class PathComparatorByPathAndPair extends WritableComparator {

		public PathComparatorByPathAndPair() {
			super(Path.class, true);
		}

		@SuppressWarnings("rawtypes")
		@Override
		public int compare(WritableComparable a, WritableComparable b) {
			Path p1 = (Path) a;
			Path p2 = (Path) b;
			return p1.compareTo(p2);
		}
	}

	public static class ReduceClass extends Reducer<Path, NounPair, NounPair, Text>{

		
		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			try{
				dpMin = Integer.parseInt(context.getConfiguration().get("dpMin","1"));
			}
			catch(Exception e){
				dpMin = 5;
			}
			System.out.println("dpMin is " + dpMin);
		}

		@Override
		protected void reduce(Path key, Iterable<NounPair> values,
				Context context)
						throws IOException, InterruptedException {			
			
			HashSet<NounPair> cached = new HashSet<NounPair>();
			NounPair last = null;
			for (NounPair nounPair : values) {
				System.out.println(nounPair);
				System.out.println("cached size is " + cached.size());
				if(cached.size() < dpMin)
				{
					System.out.println("inser " + nounPair + " to cached");
					cached.add(new NounPair(nounPair));
				}
				else
				{
					System.out.println("last is:" + last + " current is:" + nounPair);
					if(last.equals(nounPair) == false)
					{
						System.out.println("writing to context noun " + nounPair);
						context.write(nounPair, key.getPath());
					}
				}
				last = new NounPair(nounPair);
			}
			System.out.println("==================================");
			if(cached.size() == dpMin)
			{
				System.out.println("writing to context from cached");
				for (NounPair nounPair : cached) {
					System.out.println("writing to context noun " + nounPair);
					context.write(nounPair, key.getPath());
				}
			}
			
		}
	}

	public static void main(String[] args) throws IOException,
	ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		conf.set("dpMin", args[1]);
		conf.set("withInvert", args[2]);
		int numOfFiles = Integer.parseInt(args[3]);
		Job job = new Job(conf, "Final");
		job.setJarByClass(Step1.class);
		job.setMapperClass(MapClass.class);
		job.setPartitionerClass(PartitionClass.class);
		job.setReducerClass(ReduceClass.class);
		job.setMapOutputKeyClass(Path.class);
		job.setMapOutputValueClass(NounPair.class);
		job.setOutputKeyClass(NounPair.class);
		job.setOutputValueClass(Text.class);
		job.setInputFormatClass(SequenceFileInputFormat.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		FileInputFormat.addInputPaths(job, createInputPathList(numOfFiles));
		FileOutputFormat.setOutputPath(job, new org.apache.hadoop.fs.Path("s3n://ahgfinal/output/Step1/"));
		job.setSortComparatorClass(PathComparatorByPathAndPair.class);
		job.setGroupingComparatorClass(PathComparatorByPath.class);
		int completion = job.waitForCompletion(true) ? 0 : 1;
		System.exit(completion);
	}

	private static String createInputPathList(int numOfFiles) {
		StringBuilder pathList = new StringBuilder();
		for (int i = 0; i<numOfFiles ; i++ ){
			//s3://bgudsp142/syntactic-ngram/biarcs/biarcs.xx-of-99
			String fileNum = i<10 ? "0"+i : ""+i ;
			pathList.append("s3://bgudsp142/syntactic-ngram/biarcs/biarcs."+fileNum+"-of-99" );
			pathList.append(",");
		}
		pathList.deleteCharAt(pathList.lastIndexOf(","));
		return pathList.toString();
	}

}
